

package com.hazelcast.client.impl.spi.impl;

import com.hazelcast.client.config.ClientUserCodeDeploymentConfig;
import com.hazelcast.client.impl.clientside.HazelcastClientInstanceImpl;
import com.hazelcast.client.impl.protocol.ClientMessage;
import com.hazelcast.client.impl.protocol.codec.ClientDeployClassesCodec;
import com.hazelcast.internal.namespace.UserCodeNamespaceService;
import com.hazelcast.internal.nio.ClassLoaderUtil;
import com.hazelcast.jet.impl.util.ReflectionUtils;

import java.io.*;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URL;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.jar.JarEntry;
import java.util.jar.JarInputStream;

import static com.hazelcast.internal.util.EmptyStatement.ignore;

/**
 * @see UserCodeNamespaceService
 * @deprecated since 5.4, "User Code Deployment" is replaced by the "User Code Namespaces" feature
 */
@Deprecated(since = "5.4", forRemoval = true)
public class ClientUserCodeDeploymentService {
    private final ClientUserCodeDeploymentConfig clientUserCodeDeploymentConfig;
    private final ClassLoader configClassLoader;
    //List<Map.Entry> is used instead of Map to comply with generated code of client protocol
    private final List<Map.Entry<String, byte[]>> classDefinitionList = new ArrayList<>();

    public ClientUserCodeDeploymentService(ClientUserCodeDeploymentConfig clientUserCodeDeploymentConfig, ClassLoader configClassLoader) {
        this.clientUserCodeDeploymentConfig = clientUserCodeDeploymentConfig;
        this.configClassLoader = configClassLoader != null ? configClassLoader : Thread.currentThread().getContextClassLoader();
    }

    public void start() throws IOException, ClassNotFoundException {
        if (!clientUserCodeDeploymentConfig.isEnabled()) {
            return;
        }
        loadClassesFromJars();
        loadClasses();
    }

    private void loadClasses() throws ClassNotFoundException {
        for (String className : clientUserCodeDeploymentConfig.getClassNames()) {
            String resource = ReflectionUtils.toClassResourceId(className);
            try (InputStream is = configClassLoader.getResourceAsStream(resource)) {
                if (is == null) {
                    throw new ClassNotFoundException(resource);
                }
                byte[] bytes = is.readAllBytes();
                classDefinitionList.add(new AbstractMap.SimpleEntry<>(className, bytes));
            } catch (IOException e) {
                ignore(e);
            }
        }
    }

    private void loadClassesFromJars() throws IOException {
        try (ByteArrayOutputStream os = new ByteArrayOutputStream()) {
            for (String jarPath : clientUserCodeDeploymentConfig.getJarPaths()) {
                loadClassesFromJar(os, jarPath);
            }
        }
    }

    private void loadClassesFromJar(ByteArrayOutputStream os, String jarPath) throws IOException {
        try (JarInputStream inputStream = getJarInputStream(jarPath)) {
            JarEntry entry;
            do {
                entry = inputStream.getNextJarEntry();
                if (entry == null) {
                    break;
                }

                String className = ClassLoaderUtil.extractClassName(entry.getName());
                if (className == null) {
                    continue;
                }
                byte[] classDefinition = readClassDefinition(inputStream, os);
                inputStream.closeEntry();
                classDefinitionList.add(new AbstractMap.SimpleEntry<>(className, classDefinition));
            } while (true);
        }
    }

    private JarInputStream getJarInputStream(String jarPath) throws IOException {
        File file = new File(jarPath);
        if (file.exists()) {
            return new JarInputStream(new FileInputStream(file));
        }

        try {
            URL url = URI.create(jarPath).toURL();
            return new JarInputStream(url.openStream());
        } catch (IllegalArgumentException | MalformedURLException e) {
            ignore(e);
        }

        InputStream inputStream = configClassLoader.getResourceAsStream(jarPath);
        if (inputStream == null) {
            throw new FileNotFoundException("File could not be found in " + jarPath + "  and resources/" + jarPath);
        }
        return new JarInputStream(inputStream);
    }

    private byte[] readClassDefinition(JarInputStream inputStream, ByteArrayOutputStream os) throws IOException {
        os.reset();
        while (true) {
            int v = inputStream.read();
            if (v == -1) {
                break;
            }
            os.write(v);
        }
        return os.toByteArray();
    }

    public void deploy(HazelcastClientInstanceImpl client) throws ExecutionException, InterruptedException {
        if (!clientUserCodeDeploymentConfig.isEnabled() || classDefinitionList.isEmpty()) {
            return;
        }

        ClientMessage request = ClientDeployClassesCodec.encodeRequest(classDefinitionList);
        ClientInvocation invocation = new ClientInvocation(client, request, null);
        ClientInvocationFuture future = invocation.invokeUrgent();
        future.get();
    }

    //testing purposes
    public List<Map.Entry<String, byte[]>> getClassDefinitionList() {
        return classDefinitionList;
    }
}
